package com.lagou.hbase.client;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.ArrayList;

/*
 * @className demo2
 * @description
 * @author admin
 * @date 2022/7/9 15:10
 * */
public class demo2 {
    Configuration conf = null;
    Connection conn = null;
    HBaseAdmin hBaseAdmin = null;

    @Before
    public void init(){
        conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "centos7-1,centos7-2,centos7-3");
        conf.set("hbase.zookeeper.property.clientPort","2181");
        try {
            conn = ConnectionFactory.createConnection(conf);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

	// 创建表
    @Test
    public void createFriendTable(){
        try {
            hBaseAdmin = (HBaseAdmin)conn.getAdmin();
            // create table describer
            HTableDescriptor friend = new HTableDescriptor(TableName.valueOf("friend"));
            // set columns family describer
            friend.addFamily(new HColumnDescriptor("friends"));
            // perform crate
            hBaseAdmin.createTable(friend);
            System.out.println("friend table is created successfully!");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Test // 向friend表中插入数据
    public void insertDataToFriend(){
        try {
            // get a friendObject
            Table friend = conn.getTable(TableName.valueOf("friend"));
            // set rowKey
            Put uid1 = new Put(Bytes.toBytes("uid1"));
            uid1.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uid2"),Bytes.toBytes("uid2"));
            uid1.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uid5"),Bytes.toBytes("uid5"));
            uid1.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uid7"),Bytes.toBytes("uid7"));
            Put uid2 = new Put(Bytes.toBytes("uid2"));
            uid2.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uid1"),Bytes.toBytes("uid1"));
            uid2.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uid3"),Bytes.toBytes("uid3"));
            uid2.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uid7"),Bytes.toBytes("uid7"));
            Put uid3 = new Put(Bytes.toBytes("uid3"));
            uid3.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uid2"),Bytes.toBytes("uid2"));
            Put uid7 = new Put(Bytes.toBytes("uid7"));
            uid7.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uid1"),Bytes.toBytes("uid1"));
            uid7.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uid2"),Bytes.toBytes("uid2"));
            // store put
            ArrayList<Put> puts = new ArrayList<>();
            puts.add(uid1);
            puts.add(uid2);
            puts.add(uid3);
            puts.add(uid7);
            // perform insert
            friend.put(puts);
            System.out.println("data initialization succeeded!");

        } catch (IOException e) {
            e.printStackTrace();
        }
    }


	// 删除数据
    @Test
    public void deleteData(){
        try {
            // get a tableObject
            Table student = conn.getTable(TableName.valueOf("friend"));
            // get a deleteObject
            Delete delete = new Delete(Bytes.toBytes("uid1")).addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uid2"));
            // perform delete
            student.delete(delete);
            // close table
            student.close();
            // print information
            System.out.println("data is deleted successfully");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }


	// 关闭资源
    @After
    public void destroy(){
        if(conn!=null){
            try {
                conn.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        if(hBaseAdmin!=null){
            try {
                hBaseAdmin.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}
